/*
这个代码是在上一个tfidf.scala 代码基础上修正，上一个计算IDF有误，上一个就是简单的求wordcount。虽说加了wordBag过滤，
可是考虑问题出现差错，是否加入wordBag没有影响，因为所有的词都存在于wordBag 中，所以过滤等于没有过滤
*/
package spark.example

import org.apache.spark._
import SparkContext._
import java.io.File

object tfIdf {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("compute tf-idf")
    val sc = new SparkContext(conf)
    val input = args(0)
    var output = args(1)

    import scala.util.matching.Regex
    // 匹配非中文字符
    val pattern = "[^\u4e00-\u9fa5]+"r
    import scala.collection.mutable.ArrayBuffer
    // arraryBufferLine 这个可变字符数组存储目录下将所有文件内容合并成一个文件，原文件在新文件中对应于1行
    // arraryBufferLineUniq 对于新文件中每一行的单词进行去重，使得每一行的单词都唯一，便于求IDF
    val arraryBufferLineUniq = new ArrayBuffer[String]()
    val arraryBufferLine = new ArrayBuffer[String]()
    // 新文件中，行索引。只是一个标识符，没有多大的意义
    var index = 1001

    for (fd <- subDir(new File(input))) {
      // 读取每一个文件，并过滤标点符号，存储在一个RDD中
      val fileRDD = sc.textFile(fd.toString).map(x => pattern.replaceAllIn(x," "))

      // wordArray 存储一个文件里所有的单词，并过滤文件中的空白行
      val wordArray = fileRDD.flatMap(x => x.split(" ")).filter( x => x.length >0).collect()

      /* 
      将文件的所有行合并成一行的思想：
      先读取整个文件到RDD中，然后转换成数组(fileRDD.collect())，在转换成Seq，最后才是转换成字符串(wordArray.toSeq.toString)
      将此字符串插入ArrayBuffer中
      使用具体示例进一步解释这种思想
      scala> val pattern = "[^\u4e00-\u9fa5]+"r
      warning: there was one feature warning; re-run with -feature for details
      pattern: scala.util.matching.Regex = [^一-龥]+

      scala> val fileRDD = sc.textFile("/data/input/575it.seg.cln.txt").map(x => pattern.replaceAllIn(x," "))
      fileRDD: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at map at <console>:27

      scala> fileRDD.collect()
      res0: Array[String] = Array("优势 不再 评论 称 苹果 应对 软件系统 重组 升级 搜狐 ", " 搜狐 消息 北京 时间 月 日 消息 据 外国 媒体 报道 美 博客 网站 近日 发布 了 文章 称 苹果公司 应当 对 其 软件系统 进行 重组 升级 文章 内容 如下 ", " 苹果 目前 正 饱受 非议 股价 也 大幅 下跌 事实上 这种 局势 从 今年 夏季 就 已经 出现 当时       ....... 后面的省略
      scala> fileRDD.collect().toSeq
      res1: Seq[String] = WrappedArray("优势 不再 评论 称 苹果 应对 软件系统 重组 升级 搜狐 ", " 搜狐 消息 北京 时间 月 日 消息 据 外国 媒体 报道 美 博客 网站 近日 发布 了 文章 称 苹果公司 应当 对 其 软件系统 进行 重组 升级 文章 内容 如下 ", " 苹果 目前 正 饱受 非议 股价 也 大幅 下跌 事实上 这种 局势 从 今年 夏季 就 已经
      ....... 后面的省略
      scala> fileRDD.collect().toSeq.toString
      res2: String = WrappedArray(优势 不再 评论 称 苹果 应对 软件系统 重组 升级 搜狐 ,  搜狐 消息 北京 时间 月 日 消息 据 外国 媒体 报道 美 博客 网站 近日 发布 了 文章 称 苹果公司 应当 对 其 软件系统 进行 重组 升级 文章 内容 如下 ,  苹果 目前 正 饱受 非议 股价 也 大幅 下跌 事实上 这种 局势 从 今年 夏季 就 已经 出现 当时 一些
      ....... 后面的省略
      scala> fileRDD.collect().toSeq.toString.replaceAll("[^\u4e00-\u9fa5]+"," ").trim
      res3: String = 优势 不再 评论 称 苹果 应对 软件系统 重组 升级 搜狐 搜狐 消息 北京 时间 月 日 消息 据 外国 媒体 报道 美 博客 网站 近日 发布 了 文章 称 苹果公司 应当 对 其 软件系统 进行 重组 升级 文章 内容 如下 苹果 目前 正 饱受 非议 股价 也 大幅 下跌 事实上 这种 局势 从 今年 夏季 就 已经 出现 当时 一些 用户 就 抱怨 已经 过
      ....... 后面的省略
      可以看到合并成1行了，将这一字符串插入ArrayBuffer中，然后循环下去就可以将所有的文件合并成一个文件，原文件在新文件对应于一行

      感觉这个思想比较low，spark编程应该是RDD之间的相互转化(transformation类的函数)，最后使用action 类的函数计算得出结果。但是本人
      是spark初学者，在网上找了很久也没有找到适合自己需求的transformation类的函数，无赖只能借助于scala语言来解决问题

      wordArray 将这个数组转换成Set集合，去除重复的单词
      scala> val s = Array("abc","def","ghi","abc","ghi","def").toSet.toString
      s: String = Set(abc, def, ghi)  s 字符串多了Set 和()，所以需要将其过滤掉，所以有
      wordArray.toSet.toString.replaceAll("[^\u4e00-\u9fa5]+"," ") 过滤非中文字符
      最后将集合转换成字符串的目的是方便在前面加入行索引：index.toString。同时将一个完整的字符串加入ArrayBuffer中，以达到将一个
      文件合并成1行的目的
      arraryBufferLine 每一行中的单词没有去重
      arraryBufferLineUniq 每一行中的单词进行了去重，使得每一行中的单词都是唯一化。方便计算IDF，有多少行包含此单词，
      就有多少个文档包含此单词
      */ 
      val lineString = index.toString + "\t" +  wordArray.toSeq.toString.replaceAll("[^\u4e00-\u9fa5]+"," ").trim()
      val lineStringUniq = index.toString + "\t" +  wordArray.toSet.toString.replaceAll("[^\u4e00-\u9fa5]+"," ").trim()
      arraryBufferLineUniq += lineStringUniq
      arraryBufferLine += lineString
      index += 1
    }

    // IDF
    // IDF 计算公式：log(语料库中的文档总数 / (包含此单词的文档数 +1))。这里的log底数是e，自然对数e
    import scala.math._

    // output: Map(word -> value)
    val idfMap = sc.makeRDD(arraryBufferLineUniq).map(x => x.split("\t")(1)).flatMap(x => x.split(" ")).map(x => (x,1)).reduceByKey(_+_).map(x =>(x._1,log(508.toDouble / (x._2+1).toDouble))).collect().toMap


    // TF-IDF
    for(lineArray <- arraryBufferLine) {
      // tfIdrArrayBuffer 存储每一篇文章中所有单词的tf-idf 值
      val tfIdrArrayBuffer =  new ArrayBuffer[Any]()
      /* 
      lineArray.split("\t")(1) "优势 不再 评论 称 苹果 应对 软件系统 重组 升级 搜狐" 一行的内容
      lineArray.split("\t")(0) index
      tf = (文档中某一个word的词频/  该文中出现次数最多词出现的次数)
      从这里可以看出tf与具体的文档有关，具体的单词有关。一般是求文档中所有单词的tf，然后在取topN
      */
      // 对每一行求wordcount
      val lineWordCountArray = sc.makeRDD(lineArray.split("\t")(1).split(" ")).map(x => (x,1)).reduceByKey(_+_).collect()
      // 求一行词频最高值
      val maxWordCount = lineWordCountArray.map(x => x._2).max
      // lineWordCountArray = Array(("优势",10),....) x = lineWordCountArray(i) 是一个元组
      // x._1 = word  x._2 = count
      val tfIdfArray = lineWordCountArray.map(x => (x._1,(x._2.toDouble / maxWordCount.toDouble)*idfMap(x._1)))
      // lineArray.split("\t")(0) = index.toString
      
      tfIdrArrayBuffer += lineArray.split("\t")(0)
      tfIdrArrayBuffer ++= tfIdfArray
      val outputTfIdf = output + "/" + lineArray.split("\t")(0).toString
      sc.makeRDD(tfIdrArrayBuffer).saveAsTextFile(outputTfIdf)
    }
  }

  def subDir(dir :File) :Iterator[File] = {
    val dirs = dir.listFiles().filter(_.isDirectory())
    val files = dir.listFiles().filter(_.isFile())
    files.toIterator ++ dirs.toIterator
  }

}